package com.laher.demo.dao;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import com.laher.demo.model.User;

public class UserDao {
	public static final byte[] TABLE_NAME=Bytes.toBytes("user");
	public static final byte[] FAM_INFO=Bytes.toBytes("info");
	public static final byte[] TWEETS_COL=Bytes.toBytes("tweet_count");
	
	public static final byte[] COL_USER=Bytes.toBytes("user");
	public static final byte[] COL_NAME=Bytes.toBytes("name");
	public static final byte[] COL_EMAIL=Bytes.toBytes("email");
	public static final byte[] COL_PASSWORD=Bytes.toBytes("password");
	
	private Table table;
	
	public UserDao(Connection connection) throws IOException{
		this.table = connection.getTable(TableName.valueOf(TABLE_NAME));
	}
	
	private static Get mkGet(String user){
		Get get = new Get(Bytes.toBytes(user));
		get.addFamily(FAM_INFO);
		return get;
	}
	
	private static Put mkPut(UserMap u){
		Put p = new Put(Bytes.toBytes(u.user));
		p.addColumn(FAM_INFO, COL_USER, Bytes.toBytes(u.user));
		p.addColumn(FAM_INFO, COL_NAME, Bytes.toBytes(u.name));
		p.addColumn(FAM_INFO, COL_EMAIL, Bytes.toBytes(u.email));
		p.addColumn(FAM_INFO, COL_PASSWORD, Bytes.toBytes(u.password));
		return p;
	}
	
	private static Delete mkDel(String user){
		Delete del = new Delete(Bytes.toBytes(user));
		return del;
	}
	
	public void addUser(String user,String name,String email,String password) throws IOException{
		Put put = mkPut(new UserMap(user, name, email, password));
		table.put(put);
		table.close();
	}
	
	public User getUser(String user) throws IOException{
		Result result = table.get(mkGet(user));
		if(result.isEmpty()){
			return null;
		}
		UserMap u = new UserMap(result);
		table.close();
		return u;
	}
	
	public List<User> getUsers() throws IOException{
		List<User> us = new ArrayList<User>();
		ResultScanner rs = table.getScanner(new Scan());
		for (Result r : rs) {
			us.add(new UserMap(r));
		}
		return us;
	}
	
	public void deleteUser(String user) throws IOException {
		table.delete(mkDel(user));
		table.close();
	}
	
	private static class UserMap extends User{
		private UserMap(Result result) {
			this(result.getValue(FAM_INFO, COL_USER),
				result.getValue(FAM_INFO, COL_NAME),
				result.getValue(FAM_INFO, COL_EMAIL),
				result.getValue(FAM_INFO, COL_PASSWORD),
				result.getValue(FAM_INFO, TWEETS_COL)==null?
						Bytes.toBytes(0L):
						result.getValue(FAM_INFO, TWEETS_COL));
		}
		private UserMap(byte[] user, byte[] name, byte[] email, byte[] password, byte[] tweetCount) {
			this(Bytes.toString(user),Bytes.toString(name),Bytes.toString(email),Bytes.toString(password));
			this.tweetCount = Bytes.toLong(tweetCount);
		}
		private UserMap(String user,String name,String email,String password){
			this.user = user;
			this.name = name;
			this.email = email;
			this.password = password;
		}
	}
}
